package com.woniuxy.threadpool;

import com.woniuxy.untis.Sleep;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * <P> @Author: ideaAdmin
 * <P> TODO: Class
 * <P> TODO: 自定义线程池
 * <P> Date: 2022/4/10:14:23
 * <P> Description:
 */
@Slf4j(topic = "c.PoolTest01")
public class PoolTest01 {
    public static void main(String[] args) {
        ThreadPool threadPool = new ThreadPool(1, 2,
                TimeUnit.SECONDS, 1,
                (RejectPolicy<Runnable>) (queue, task) -> {

                    //TODO
                    // 1. 队列满了死等
                    // queue.put(task);
                    // 2. 带超时等待
                    //   queue.offer(task,1500,TimeUnit.MILLISECONDS);
                    // 3. 放弃任务执行
                    //  log.debug("放弃执行:---{}",task);
                    // 4. 抛出异常
                    //  throw new RuntimeException("队列已满 不能执行:" + task);
                    // 5. 让调用者自己执行任务
                    task.run();
                });

        for (int i = 0; i < 4; i++) {
            int j = i;
            threadPool.execute(() -> {
                Sleep.sleep(1000);
                log.debug("{}", j);
            });
        }
    }
}

//TODO 拒绝策略
@FunctionalInterface
interface RejectPolicy<T> {
    /**
     * TODO:自定义策略
     **/
    public void reject(BlockingQueue1<T> queue, T task);


}

@Slf4j(topic = "c.ThreadPool")
class ThreadPool<T> {
    /**
     * TODO 任务队列
     */
    private BlockingQueue1<Runnable> taskQueue;

    /**
     * TODO:线程集合
     **/
    private HashSet<Worker> workers = new HashSet();


    /**
     * TODO:核心线程数
     **/
    private int coreSize;

    /**
     * TODO:获取任务超时时间
     **/

    private long timeout;


    private TimeUnit timeUnit;

    /**
     * TODO 拒绝策略接口
     */
    private RejectPolicy<Runnable> rejectPolicy;


    public ThreadPool(int coreSize, long timeout,
                      TimeUnit timeUnit, int queueCapacity,
                      RejectPolicy<Runnable> rejectPolicy) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue = new BlockingQueue1<>(queueCapacity);
        this.rejectPolicy = rejectPolicy;
    }

    /**
     * TODO:执行任务
     **/
    public void execute(Runnable task) {
        //TODO 任务数量没有超过coresize 直接交给worker对象执行
        // 如果超过了 放入任务队列暂存
        synchronized (workers) {
            if (workers.size() < coreSize) {
                Worker worker = new Worker(task);
                log.debug("新值worker:{},任务对象：{}", worker, task);
                workers.add(worker);
                worker.start();
            } else {

                // taskQueue.put(runnable);
                //TODO
                // 1. 队列满了死等
                // 2. 带超时等待
                // 3. 放弃任务执行
                // 4. 抛出异常
                // 5. 让调用者自己执行任务
                taskQueue.tryPut(rejectPolicy, task);
            }
        }
    }

    class Worker extends Thread {
        private Runnable task;

        public Worker(Runnable task) {
            this.task = task;
        }

        @Override
        public void run() {
            //TODO 执行任务
            //  1.task为空 执行任务
            //  2.task执行完毕 在从任务队列获取任务并执行
            while (task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {
                try {
                    log.debug("正在执行：{}", task);
                    task.run();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    task = null;
                }
            }
            synchronized (workers) {
                log.debug("worder被移除：{}", this);
                workers.remove(this);
            }
        }
    }
}

@Slf4j(topic = "c.BlockingQueue1")
class BlockingQueue1<T> {
    /**
     * TODO 任务队列 双向链表
     */
    private Deque<T> queue = new ArrayDeque();

    /**
     * TODO 锁
     */
    private ReentrantLock lock = new ReentrantLock();

    /**
     * TODO 生产者的条件变量 如果队列满了 则不能再放入
     */
    private Condition fullWaitSet = lock.newCondition();

    /**
     * TODO 消费者的阻塞条件变量 如果队列为空 不能继续消息
     */
    private Condition emptyWaitSet = lock.newCondition();


    /**
     * TODO 队列容量
     */

    private int capacity;

    public BlockingQueue1(int capacity) {
        this.capacity = capacity;
    }


    /**
     * TODO:带超时的阻塞获取
     **/
    public T poll(long timeOut, TimeUnit timeUnit) {

        lock.lock();
        try {
            //将超时时间统一转为  纳秒
            long nanos = timeUnit.toNanos(timeOut);

            //TODO 如果为空 需要等待
            while (queue.isEmpty()) {
                try {

                    if (nanos <= 0) {
                        return null;
                    }
                    //TODO 返回的是剩余需要等待的时间
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //TODO 取出头部
            T t = queue.removeFirst();
            //TODO 唤醒生产者生成
            fullWaitSet.signal();
            return t;
        } finally {
            lock.unlock();
        }

    }

    /**
     * TODO:阻塞获取
     **/
    public T take() {
        lock.lock();
        try {
            //TODO 如果为空 需要等待
            while (queue.isEmpty()) {
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //TODO 取出头部
            T t = queue.removeFirst();
            //TODO 唤醒生产者生成
            fullWaitSet.signal();
            return t;
        } finally {
            lock.unlock();
        }
    }

    /**
     * TODO: 阻塞添加
     **/
    public void put(T task) {
        lock.lock();
        try {

            while (queue.size() == capacity) {
                try {
                    log.debug("等待加入任务队列:{}", task);
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("加入到任务队列，{}", task);
            queue.addLast(task);
            emptyWaitSet.signal();
        } finally {
            lock.unlock();
        }
    }

    /**
     * TODO:带超时时间的阻塞添加
     **/
    public boolean offer(T task, long timeout, TimeUnit timeUnit) {
        lock.lock();
        try {
            long nanos = timeUnit.toNanos(timeout);
            while (queue.size() == capacity) {
                try {
                    log.debug("等待加入任务队列:{}", task);

                    if (nanos <= 0) {
                        log.debug("任务超时,抛弃,{}", task);
                        return false;
                    }
                    nanos = fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("加入到任务队列，{}", task);
            queue.addLast(task);
            //TODO 唤醒消费者线程消费
            emptyWaitSet.signal();
            return true;
        } finally {
            lock.unlock();
        }
    }


    /**
     * TODO:获取队列大小
     **/
    public int size() {
        lock.lock();
        try {
            return queue.size();
        } finally {
            lock.unlock();
        }
    }

    public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
        lock.lock();

        try {
            //TODO 队列是否已满
            if (queue.size() == capacity) {
                rejectPolicy.reject(this, task);
            } else {
                log.debug("加入到任务队列，{}", task);
                queue.addLast(task);
                //TODO 唤醒消费者线程消费
                emptyWaitSet.signal();
            }
        } finally {
            lock.unlock();
        }
    }
}
